package rabbitmq

import (
	"fmt"

	"github.com/golang/protobuf/proto"
	"github.com/streadway/amqp"

	//"golang.org/x/sys/windows"
	"sync"
	"time"
	"unsafe"
)

// 定义生产者接口
type Producer interface {
	MsgContent() string
}

// 定义接收者接口
type Receiver interface {
	Consumer([]byte) error
}

// 定义RabbitMQ对象
type RabbitMQ struct {
	conn         *amqp.Connection
	channel      *amqp.Channel
	queueName    string // 队列名称
	routingKey   string // key名称
	exchangeName string // 交换机名称
	exchangeType string // 交换机类型
	rabbitUrl    string // 地址
	producerList []Producer
	receiverList []Receiver
	mu           sync.RWMutex
}

// 定义队列交换机对象
type QueueExchange struct {
	QuName string // 队列名称
	RtKey  string // key值
	ExName string // 交换机名称
	ExType string // 交换机类型
}

// 链接rabbitMQ
func (r *RabbitMQ) mqConnect() {
	var err error
	r.conn, err = amqp.Dial(r.rabbitUrl)

	if err != nil {
		fmt.Println("MQ打开链接失败:", err, "url:", r.rabbitUrl)
	}

	r.channel, err = r.conn.Channel()

	if err != nil {
		fmt.Println("MQ打开管道失败:", err, "url:", r.rabbitUrl)
	}
}

// 关闭RabbitMQ连接
func (r *RabbitMQ) mqClose() {

	if r.channel == nil {
		return
	}
	// 先关闭管道,再关闭链接
	err := r.channel.Close()
	if err != nil {
		fmt.Printf("MQ管道关闭失败:%s \n", err)
	}
	err = r.conn.Close()
	if err != nil {
		fmt.Printf("MQ链接关闭失败:%s \n", err)
	}
}

// 创建一个新的操作对象
func NewMQ(q *QueueExchange) *RabbitMQ {
	return &RabbitMQ{
		queueName:    q.QuName,
		routingKey:   q.RtKey,
		exchangeName: q.ExName,
		exchangeType: q.ExType,
	}
}

func NewMQSender(rtKey, exName, exType string) *RabbitMQ {
	return &RabbitMQ{
		queueName:    "test",
		routingKey:   rtKey,
		exchangeName: exName,
		exchangeType: exType,
	}
}

// 启动RabbitMQ客户端,并初始化
func (r *RabbitMQ) Start(url string) {
	r.rabbitUrl = url
	// 开启监听生产者发送任务
	for _, producer := range r.producerList {
		r.listenProducer(producer)
	}
	/*
		go func (r *RabbitMQ){
			// 开启监听接收者接收任务
			for _, receiver := range r.receiverList {
				r.listenReceiver(receiver)
			}
		}(r)
	*/
	time.Sleep(1 * time.Second)
}

func (r *RabbitMQ) Stop() {
	r.mqClose()
}

// 注册发送指定队列指定路由的生产者
func (r *RabbitMQ) RegisterProducer(producer Producer) {
	r.producerList = append(r.producerList, producer)
}

// 发送任务
func (r *RabbitMQ) listenProducer(producer Producer) {

	//fmt.Println("pre send msg:", producer.MsgContent())
	// 验证链接是否正常,否则重新链接
	if r.channel == nil {
		r.mqConnect()
	}

	if r.conn == nil || r.channel == nil {
		fmt.Println("can't connect rabbitmq.")
		return
	}

	err := r.channel.ExchangeDeclare(r.exchangeName, r.exchangeType, false, false, false, false, nil)

	if err != nil {
		fmt.Println("exchange declare fail:", err, "param:", r.exchangeName, r.exchangeName)
		return
	}

	// 发送任务消息
	err = r.channel.Publish(r.exchangeName, r.routingKey, false, false, amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte(producer.MsgContent()),
	})
	if err != nil {
		fmt.Printf("MQ任务发送失败:%s \n", err)
		return
	}
}

// 注册接收指定队列指定路由的数据接收者
func (r *RabbitMQ) RegisterReceiver(receiver Receiver) {
	r.mu.Lock()
	r.receiverList = append(r.receiverList, receiver)
	r.mu.Unlock()
}

// 监听接收者接收任务
func (r *RabbitMQ) listenReceiver(receiver Receiver) {
	fmt.Println("enter reciver.")
	// 处理结束关闭链接
	//defer r.mqClose()
	// 验证链接是否正常
	if r.channel == nil {
		fmt.Println("rabbitmq connect url:", r.rabbitUrl)
		r.mqConnect()
	}

	err := r.channel.ExchangeDeclare(r.exchangeName, r.exchangeType, false, false, false, false, nil)

	if err != nil {
		fmt.Println("exchange declare fail:", err, "param:", r.exchangeName, r.exchangeType)
		return
	}

	q, err := r.channel.QueueDeclare(r.queueName, true, false, false, false, nil)

	if err != nil {
		fmt.Println("queue declare fail:", err)
		return
	}

	// 绑定任务
	err = r.channel.QueueBind(q.Name, r.routingKey, r.exchangeName, true, nil)
	if err != nil {
		fmt.Printf("绑定队列失败:%s \n", err)
		return
	} else {
		fmt.Println("success bind queue:", q.Name, r.routingKey, r.exchangeName)
	}
	// 获取消费通道,确保rabbitMQ一个一个发送消息
	//err = r.channel.Qos(1, 0, true)
	msgList, err := r.channel.Consume(r.queueName, "", false, false, false, false, nil)
	if err != nil {
		fmt.Printf("receiver获取消费通道异常:%s \n", err)
		return
	}
	for msg := range msgList {
		// 处理数据
		err := receiver.Consumer(msg.Body)
		if err != nil {
			err = msg.Ack(true)
			if err != nil {
				fmt.Printf("确认消息未完成异常:%s \n", err)
				return
			}
		} else {
			// 确认消息,必须为false
			err = msg.Ack(false)
			if err != nil {
				fmt.Printf("确认消息完成异常:%s \n", err)
				return
			}
			//return
		}
	}
}

func (r *RabbitMQ) SendMessage(topic string, funCode uint32, sid uint32, data proto.Message) {
	msg, err := proto.Marshal(data)
	if err != nil {
		fmt.Println("Marshal msg fail:", err)
		return
	}

	var sendBuf []byte = make([]byte, len(msg)+8)

	// 赋值funcode
	{
		p := unsafe.Pointer(&funCode)
		q := (*[4]byte)(p)
		copy(sendBuf[0:], (*q)[0:])
	}

	// 赋值sid
	{
		p := unsafe.Pointer(&sid)
		q := (*[4]byte)(p)
		copy(sendBuf[4:], (*q)[0:])
	}

	copy(sendBuf[8:], msg)
	r.PublishMessage(topic, string(sendBuf))
	fmt.Printf("send message, funCode[%d] sid[%d] \n %s\n", funCode, sid, data.String())
}

func (r *RabbitMQ) PublishMessage(topic string, msg string) {
	t := &MiMessage{
		msg,
	}
	r.routingKey = topic
	go r.listenProducer(t)
}

/****************************rabbitmq******************/

type MiMessage struct {
	msgContent string
}

// 实现发送者
func (t *MiMessage) MsgContent() string {
	return t.msgContent
}

// 实现接收者
func (t *MiMessage) Consumer(dataByte []byte) error {
	fmt.Println( /*windows.GetCurrentThreadId(),*/ "receive msg:", string(dataByte))
	return nil
}
